package cn.com.klun.tjy.other.zkUsed;

import cn.com.klun.tjy.other.util.ExecutorServiceUtil;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;

import org.apache.zookeeper.ZooDefs.*;

public class Worker implements Watcher {
    private Logger logger = LoggerFactory.getLogger(Worker.class);

    private ExecutorService executorService = ExecutorServiceUtil.newExecutorInstance();
    Random random = new Random();
    ZooKeeper zk;
    String hostPort;
    String serverId = Integer.toHexString(random.nextInt());
    private String status ;
    private TaskCache taskCache;
    Worker(String hostPort){
        this.hostPort = hostPort;
    }
    void startZk() throws IOException {
        zk = new ZooKeeper(hostPort,15000,this);
    }
    void stopZK() throws InterruptedException {
        zk.close();
    }
    @Override
    public void process(WatchedEvent event) {
        logger.info(event.toString()+","+hostPort);
    }
    void register(){
        zk.create("/workers/worker-"+serverId,
                "Idle".getBytes(),
                Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL,
                createWorkerCallBack,null);
    }

    AsyncCallback.StringCallback createWorkerCallBack = new AsyncCallback.StringCallback() {
        @Override
        public void processResult(int i, String s, Object o, String s1) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    register(); //注册工作节点
                    break;
                case OK:
                    logger.info("节点注册成功+"+serverId);
                    break;
                case NODEEXISTS:
                    logger.warn("已经存在该节点"+serverId);
                    break;
                default:
                    logger.error("运行异常："+KeeperException.create(KeeperException.Code.get(i),s));
            }
        }
    };
    // 节点状态 回调
    AsyncCallback.StatCallback statUpdateCallback = new AsyncCallback.StatCallback() {
        @Override
        public void processResult(int i, String s, Object o, Stat stat) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    updateStatus((String)o);//更新状态重试
                    return;
                case OK:
                    logger.info("***#####**看一下是不是workers的路径: "+s);
                    logger.info("##########状态更新成功："+(String)o);
                    setStatus((String)o); //更新成功 放入status whm
            }
        }
    };
    // 设置监视点 异步，有变动则改变
    synchronized private void updateStatus(String status){
        logger.info("*********对比+"+this.status+"::"+status);
        if(status != this.status){
            // -1 禁止版本号检查
            //zk.setData("/workers/"+name,status.getBytes(),-1,statUpdateCallback,status);
            zk.setData("/workers/worker-"+serverId,status.getBytes(),-1,statUpdateCallback,status);
        }
    }
    void setStatus(String status){
        this.status = status;
        //updateStatus(status);
    }

    /**
     * 开启单独的线程对从节点的已分配任务进行循环
     */
    Watcher newTaskWatcher = new Watcher() {
        @Override
        public void process(WatchedEvent watchedEvent) {
            if(watchedEvent.getType() == Event.EventType.NodeChildrenChanged){
                assert new String("/assign/worker-"+serverId).equals(watchedEvent.getPath());
                getTask();//当收到子节点的变化的通知后，获得子节点的列表
            }
        }
    };

    /**
     * 获取 worker 下的任务节点
     */
    void getTask(){
        zk.getChildren("/assign/worker-"+serverId,newTaskWatcher,tasksGetChildrenCallback,null);
    }
    AsyncCallback.ChildrenCallback tasksGetChildrenCallback = new AsyncCallback.ChildrenCallback() {
        @Override
        public void processResult(int i, String s, Object o, List<String> children) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    getTask();
                    break;
                case OK:
                    if(children != null){
                        executorService.execute(new Runnable() {
                            List<String> children;
                            AsyncCallback.DataCallback cb;
                            public Runnable init(List<String> children,AsyncCallback.DataCallback cb){
                                this.children = children;
                                this.cb = cb;
                                return this;
                            }
                            @Override
                            public void run() {
                                logger.info("looping into tasks");
                                synchronized (taskCache.getOnGingTask()){
                                    for(String task : children){
                                        if(!(taskCache.getOnGingTask().contains(task))){
                                            logger.trace("new task:{}",task);
                                            zk.getData("/assign/worker-"+serverId+"/"+task,false,cb,task);
                                            taskCache.getOnGingTask().add(task);
                                        }
                                    }
                                }
                            }
                        }.init(children,taskDataCallback));
                    }
                    break;
                default:
                    System.out.println("失败");

            }
        }
    };
    AsyncCallback.DataCallback taskDataCallback = new AsyncCallback.DataCallback() {
        @Override
        public void processResult(int i, String s, Object o, byte[] bytes, Stat stat) {
            switch (KeeperException.Code.get(i)){
                case CONNECTIONLOSS:
                    break;
                case OK:
                    //处理返回的新任务数据
                    break;
                default:
                    logger.error("异常");
            }
        }
    };

    public static void main(String[] args) throws IOException, InterruptedException {
        Worker w  = new Worker("192.168.137.11:2181");
        w.startZk();
        w.register();
        Thread.sleep(3000000);
        w.stopZK();
    }
}
